package org.sunyaxing.transflow.transflowapp.plugins.inputs;

import com.alibaba.fastjson2.JSONObject;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sunyaxing.transflow.api.HandleData;
import org.sunyaxing.transflow.api.TransData;
import org.sunyaxing.transflow.api.common.Handle;
import org.sunyaxing.transflow.api.common.ano.Extension;
import org.sunyaxing.transflow.api.common.ano.HandleItem;
import org.sunyaxing.transflow.api.common.ano.JobParamItem;
import org.sunyaxing.transflow.api.common.ano.ScopeContentCheck;
import org.sunyaxing.transflow.api.extensions.base.ExtensionContext;
import org.sunyaxing.transflow.api.extensions.base.InputUtil;
import org.sunyaxing.transflow.api.extensions.base.types.TransFlowInput;
import org.sunyaxing.transflow.transflowapp.plugins.inputs.nettyinput.TransflowChannelInitializer;

import java.util.List;
import java.util.function.Function;

@Extension("plugin-netty-demo-input")
@ScopeContentCheck(value = {
        @JobParamItem(field = "host", label = "HOST", defaultValue = "0.0.0.0"),
        @JobParamItem(field = "port", label = "端口", defaultValue = "8080"),
},handle = @HandleItem(field = "url", label = "接口"))
public class NettyInputExtension extends TransFlowInput<String, String> {

    private static final Logger log = LoggerFactory.getLogger(NettyInputExtension.class);
    private EventLoopGroup boss;
    private EventLoopGroup worker;
    private ChannelFuture future;

    public NettyInputExtension(ExtensionContext extensionContext) {
        super(extensionContext);
    }

    @Override
    public void commit(HandleData<String> handleData) {

    }

    @Override
    protected void afterInitHandler(JSONObject config, List<Handle> handles) {
        String host = config.getString("host");
        Integer port = config.getInteger("port");
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        // 做是否支持epoll轮询判断以获取更高性能 EpollEventLoopGroup LINUX
        boss = Epoll.isAvailable() ? new EpollEventLoopGroup(1) : new NioEventLoopGroup(1);
        worker = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();
        try {
            serverBootstrap.group(boss, worker)
                    .channel(Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                    .childHandler(new TransflowChannelInitializer(new InputUtil<>(this)))
                    .childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(131072))
                    .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(32 * 1024, 128 * 1024))
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childOption(ChannelOption.ALLOCATOR, new PooledByteBufAllocator(true))
                    .childOption(ChannelOption.TCP_NODELAY, true);

            future = serverBootstrap.bind(host, port).sync();
            future.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> {
                channelFuture.channel().close();
            });
        } catch (Exception e) {
            log.error("e: ", e);
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        } finally {
        }
    }

    @Override
    public Function<TransData<String>, String> parseHandleToConsumer(String handleId, String handleValue) {
        return TransData::getData;
    }

    @Override
    public void destroy() {
        try {
            future.channel().close().sync();
        } catch (Exception e) {
            log.error("关闭netty错误", e);
        } finally {
            if (boss != null) {
                this.boss.shutdownGracefully();
            }
            if (worker != null) {
                this.worker.shutdownGracefully();
            }
        }
    }
}
